package com.inspur.icity.bus.base.core.repo;

import com.google.common.base.Strings;
import net.spy.memcached.*;
import net.spy.memcached.auth.AuthDescriptor;
import net.spy.memcached.auth.PlainCallbackHandler;
import net.spy.memcached.metrics.MetricCollector;
import net.spy.memcached.metrics.MetricType;
import net.spy.memcached.ops.Operation;
import net.spy.memcached.ops.OperationQueueFactory;
import net.spy.memcached.protocol.binary.BinaryOperationFactory;
import net.spy.memcached.transcoders.Transcoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;

/**
 * Created by limiemie on 14-9-3.
 */
public class AliyunConnectionFactory extends DefaultConnectionFactory {
    Logger logger = LoggerFactory.getLogger(getClass());

    private static final String SYSTEM_PROPERTY_MEMCACHED_PROPERTIES_FILENAME = "memcached.properties.filename";

    /**
     *
     */
    private static final String MEMCACHED_RESOURCE = "memcached.properties";

    protected OperationQueueFactory opQueueFactory;
    protected OperationQueueFactory readQueueFactory;
    protected OperationQueueFactory writeQueueFactory;

    protected Transcoder<Object> transcoder;

    protected FailureMode failureMode;

    protected Collection<ConnectionObserver> initialObservers =
            Collections.emptyList();

    protected OperationFactory opFact = new BinaryOperationFactory();

    protected Locator locator = Locator.ARRAY_MOD;
    protected long opTimeout = -1;
    protected boolean isDaemon = false;
    protected boolean shouldOptimize = false;
    protected boolean useNagle = false;
    protected long maxReconnectDelay =
            DefaultConnectionFactory.DEFAULT_MAX_RECONNECT_DELAY;

    protected int readBufSize = -1;
    protected HashAlgorithm hashAlg;
    protected AuthDescriptor authDescriptor = null;
    protected long opQueueMaxBlockTime = -1;

    protected int timeoutExceptionThreshold =
            DefaultConnectionFactory.DEFAULT_MAX_TIMEOUTEXCEPTION_THRESHOLD;

    protected MetricType metricType = null;
    protected MetricCollector collector = null;
    protected ExecutorService executorService = null;
    protected long authWaitTime = DefaultConnectionFactory.DEFAULT_AUTH_WAIT_TIME;

    public AliyunConnectionFactory(){
        logger.debug("init AliyunConnectionFactory...");

        String memcachedPropertiesFilename = System.getProperty(SYSTEM_PROPERTY_MEMCACHED_PROPERTIES_FILENAME, MEMCACHED_RESOURCE);
        Properties config = new Properties();

        // load the properties specified from /memcached.properties, if present
        InputStream input = getClass().getClassLoader()
                .getResourceAsStream(memcachedPropertiesFilename);
        if (input != null) {
            try {
                config.load(input);
            } catch (IOException e) {
                throw new RuntimeException("An error occurred while reading classpath property '"
                        + memcachedPropertiesFilename
                        + "', see nested exceptions", e);
            } finally {
                try {
                    input.close();
                } catch (IOException e) {
                    // close quietly
                }
            }
        }

        String userName = config.getProperty("org.mybatis.caches.memcached.username");
        String password = config.getProperty("org.mybatis.caches.memcached.password");

        if(!Strings.isNullOrEmpty(userName)){
            authDescriptor = new AuthDescriptor(new String[]{"PLAIN"},
                    new PlainCallbackHandler(userName, password));
            logger.debug("username: {}, password: {}", userName,password);
        }
    }

    @Override
    public BlockingQueue<Operation> createOperationQueue() {
        return opQueueFactory == null ? super.createOperationQueue()
                : opQueueFactory.create();
    }

    @Override
    public BlockingQueue<Operation> createReadOperationQueue() {
        return readQueueFactory == null ? super.createReadOperationQueue()
                : readQueueFactory.create();
    }

    @Override
    public BlockingQueue<Operation> createWriteOperationQueue() {
        return writeQueueFactory == null ? super.createReadOperationQueue()
                : writeQueueFactory.create();
    }

    @Override
    public NodeLocator createLocator(List<MemcachedNode> nodes) {
        switch (locator) {
            case ARRAY_MOD:
                return new ArrayModNodeLocator(nodes, getHashAlg());
            case CONSISTENT:
                return new KetamaNodeLocator(nodes, getHashAlg());
            default:
                throw new IllegalStateException("Unhandled locator type: " + locator);
        }
    }

    @Override
    public Transcoder<Object> getDefaultTranscoder() {
        return transcoder == null ? super.getDefaultTranscoder() : transcoder;
    }

    @Override
    public FailureMode getFailureMode() {
        return failureMode == null ? super.getFailureMode() : failureMode;
    }

    @Override
    public HashAlgorithm getHashAlg() {
        return hashAlg == null ? super.getHashAlg() : hashAlg;
    }

    @Override
    public Collection<ConnectionObserver> getInitialObservers() {
        return initialObservers;
    }

    @Override
    public OperationFactory getOperationFactory() {
        return opFact == null ? super.getOperationFactory() : opFact;
    }

    @Override
    public long getOperationTimeout() {
        return opTimeout == -1 ? super.getOperationTimeout() : opTimeout;
    }

    @Override
    public int getReadBufSize() {
        return readBufSize == -1 ? super.getReadBufSize() : readBufSize;
    }

    @Override
    public boolean isDaemon() {
        return isDaemon;
    }

    @Override
    public boolean shouldOptimize() {
        return shouldOptimize;
    }

    @Override
    public boolean useNagleAlgorithm() {
        return useNagle;
    }

    @Override
    public long getMaxReconnectDelay() {
        return maxReconnectDelay;
    }

    @Override
    public AuthDescriptor getAuthDescriptor() {
        return authDescriptor;
    }

    @Override
    public long getOpQueueMaxBlockTime() {
        return opQueueMaxBlockTime > -1 ? opQueueMaxBlockTime
                : super.getOpQueueMaxBlockTime();
    }

    @Override
    public int getTimeoutExceptionThreshold() {
        return timeoutExceptionThreshold;
    }

    @Override
    public MetricType enableMetrics() {
        return metricType == null ? super.enableMetrics() : metricType;
    }

    @Override
    public MetricCollector getMetricCollector() {
        return collector == null ? super.getMetricCollector() : collector;
    }

    @Override
    public ExecutorService getListenerExecutorService() {
        return executorService == null ? super.getListenerExecutorService() : executorService;
    }

    @Override
    public boolean isDefaultExecutorService() {
        return executorService == null;
    }

    @Override
    public long getAuthWaitTime() {
        return authWaitTime;
    }

    /**
     * Type of protocol to use for connections.
     */
    public static enum Protocol {
        /**
         * Use the text (ascii) protocol.
         */
        TEXT,
        /**
         * Use the binary protocol.
         */
        BINARY
    }

    /**
     * Type of node locator to use.
     */
    public static enum Locator {
        /**
         * Array modulus - the classic node location algorithm.
         */
        ARRAY_MOD,
        /**
         * Consistent hash algorithm.
         *
         * This uses ketema's distribution algorithm, but may be used with any hash
         * algorithm.
         */
        CONSISTENT,
        /**
         * VBucket support.
         */
        VBUCKET
    }
}
